package net.bwie.vehicle.dws.job1;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.guanjuntao.util.JdbcUtil;
import net.bwie.realtime.guanjuntao.util.KafkaUtil;
import net.bwie.vehicle.dws.bean1.VehicleData;
import net.bwie.vehicle.dws.function1.VehicleMonitoringFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
/*
  实时监控车辆状态
 */
public class VehicleMonitoringJob {
    public static void main(String[] args) throws Exception {
        // TODO: 2025/6/7 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO: 2025/6/7 拉取kafka数据
        DataStream<String> stringDataStream = KafkaUtil.consumerKafka(env, "car-vehicle-data");
        // TODO: 2025/6/7 转换为实体类
        SingleOutputStreamOperator<VehicleData> vehicleDataMap = stringDataStream.map(o -> JSON.parseObject(o, VehicleData.class));
        DataStream<VehicleData> resultStream = vehicleDataMap
                .keyBy(VehicleData::getVin)
                .process(new VehicleMonitoringFunction());
        // TODO: 2025/6/7 保存数据
        SingleOutputStreamOperator<String> outPutMap = resultStream.map(new MapFunction<VehicleData, String>() {
            @Override
            public String map(VehicleData value) throws Exception {
                String output = value.getVin() + ',' +
                        value.getSpeed() + ',' +
                        value.getMileage() + ',' +
                        value.getBatteryLevel() + ',' +
                        value.getBatteryTemp() + ',' +
                        value.getMotorTemp() + ',' +
                        value.getChargingStatus() + ',' +
                        value.getEnergyConsumption();
                return output;
            }
        });
        outPutMap.print();
        String upsertSql = "INSERT INTO new_car.dws_vehiclemonitorin (vin, speed, mileage, batteryLevel, batteryTemp, motorTemp, chargingStatus, energyConsumption)\n" +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?)";
        // TODO: 2025/5/22 插入数据
        JdbcUtil.sinkToClickhouseUpsert(outPutMap, upsertSql);
        // TODO: 2025/6/7 触发执行
        env.execute("VehicleMonitoringJob");
    }
}
